package rocketmq

import (
	"context"
	"sync"
	"time"

	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/aliyun"

	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/apache"
	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"

	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

var (
	onceDo sync.Once
	eng    *SubscribeEngine
)

// HandlerFunc defines the handler used by rocketmq middleware as return value.
type HandlerFunc func(*Context)

// HandlersChain defines a HandlerFunc array.
type HandlersChain []HandlerFunc

// Last returns the last handler in the chain. the last handler is the main one.
func (c HandlersChain) Last() HandlerFunc {
	if length := len(c); length > 0 {
		return c[length-1]
	}
	return nil
}

type SubscribeEngine struct {
	logger log.Logger

	conf config.Config

	rocketMqType string

	client internal.RocketProxy

	handlers HandlersChain

	subList []*Subscriber

	subTransList []*Subscriber

	wg sync.WaitGroup

	ctx    context.Context
	cancel context.CancelFunc
}

type SubscribeEngineOption func(*SubscribeEngine)

func NewSubscribeEngine(options ...SubscribeEngineOption) *SubscribeEngine {
	onceDo.Do(func() {
		eng = &SubscribeEngine{}
		eng.wg = sync.WaitGroup{}

		ctx, cancel := context.WithCancel(context.Background())
		eng.ctx = ctx
		eng.cancel = cancel

		for _, option := range options {
			option(eng)
		}

		if eng.conf == nil {
			eng.conf = config.NewNullConfig()
		}

		if eng.logger == nil {
			eng.logger = log.NewLogger()
		}

		if eng.handlers == nil {
			eng.handlers = make(HandlersChain, 0)
		}

		if eng.subList == nil {
			eng.subList = make([]*Subscriber, 0)
		}

		if eng.subTransList == nil {
			eng.subTransList = make([]*Subscriber, 0)
		}

		eng.rocketMqType = eng.conf.GetString("rocket_mq_type")
		if eng.rocketMqType == internal.RocketMqApache {
			/*apache 初始化*/
			var cliOpt apache.MQClientOptions
			eng.client = apache.NewMQClient(
				cliOpt.WithLogger(eng.logger),
				cliOpt.WithConf(eng.conf))
		} else {
			/*aliyun 版本初始化*/
			var cliOpt aliyun.MQClientOptions
			eng.client = aliyun.NewMQClient(
				cliOpt.WithLogger(eng.logger),
				cliOpt.WithConf(eng.conf))
		}
	})

	return eng
}

func WithSubscribeEngineConf(conf config.Config) SubscribeEngineOption {
	return func(se *SubscribeEngine) {
		se.conf = conf
	}
}

func WithSubscribeEngineLogger(logger log.Logger) SubscribeEngineOption {
	return func(se *SubscribeEngine) {
		se.logger = logger
	}
}

func WithSubscribeHandlersChain(handlers ...HandlerFunc) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.handlersChain = handlers
	}
}

func (se *SubscribeEngine) Use(middleware ...HandlerFunc) {
	se.handlers = append(se.handlers, middleware...)
}

func (se *SubscribeEngine) combineHandlers(handlers HandlersChain) HandlersChain {
	finalSize := len(se.handlers) + len(handlers)
	if finalSize >= int(abortIndex) {
		panic("too many handlers")
	}
	mergedHandlers := make(HandlersChain, finalSize)
	copy(mergedHandlers, se.handlers)
	copy(mergedHandlers[len(se.handlers):], handlers)
	return mergedHandlers
}

func (se *SubscribeEngine) Start() {
	if len(se.subList) == 0 {
		se.logger.Panicf("没有订阅任何信息!!!")
	}

	if se.rocketMqType == internal.RocketMqApache {
		se.startApache()
	} else {
		// 消费队列
		for _, subscribe := range se.subList {
			for i := 0; i < subscribe.concurrency; i++ {
				se.wg.Add(1)
				go func(sub *Subscriber) {
					se.logger.Infof("RocketMqAliyun订阅信息 topicName[%s] groupID[%s] messageTag[%s]", sub.topicName, sub.groupID, sub.messageTag)
					for {
						select {
						case <-se.ctx.Done():
							se.logger.Infof("close the rocketmq subscribe")
							se.wg.Done()
							return
						default:
						}

						ctx, cancel := context.WithTimeout(se.ctx, 35*time.Second)
						sub.subscribeAliyun(ctx)
						cancel()
					}
				}(subscribe)
			}
		}
	}

	se.logger.Infof("SubscribeEngine Start Success!")
}

func (se *SubscribeEngine) startApache() {
	// 按groupId 合并Subscribe
	subGroupInf := map[string]map[string]*Subscriber{}
	for _, s := range se.subList {
		key := s.groupID
		if key == "" {
			key = internal.DefaultApacheGroupId
		}
		if _, ok := subGroupInf[key]; !ok {
			subGroupInf[key] = map[string]*Subscriber{}
		}
		if _, ok := subGroupInf[key][s.topicName]; ok {
			se.logger.Panicf("groupId[%s] topicName[%s] 重复注册；", key, s.topicName)
		}
		subGroupInf[key][s.topicName] = s
	}

	var c internal.MQConsumer
	var err error

	// 按groupId 遍历, 每个groupId 创建一个Consumer， 订阅topic
	for groupID, v := range subGroupInf {
		if groupID == internal.DefaultApacheGroupId {
			continue
		}

		first := true
		for _, sub := range v {
			if first {
				// 创建Consumer
				c, err = sub.client.CreateConsumer(sub.topicName, groupID, sub.messageTag)
				if err != nil {
					se.logger.Panicf("startApache:CreateConsumer[%s][%s][%s] err[%s]", sub.topicName, groupID, sub.messageTag, err)
					return
				}
				se.logger.Infof("groupId [%s] consumer 创建成功;", groupID)
				first = false
			}

			// 注册Subscriber
			err = sub.subscribeApache(c)
			if err != nil {
				sub.logger.Panicf("startApache: subscribeApache [%s][%s][%s] err[%s]", sub.topicName, groupID, sub.messageTag, err)
				return
			}
		}

		// Start consumer
		err = c.Start()
		if err != nil {
			se.logger.Panicf("startApache: Apache consumer start err[%s]", err)
			return
		}
	}

	// 处理无groupId情况
	if v, ok := subGroupInf[internal.DefaultApacheGroupId]; ok {
		groupID := ""
		for _, sub := range v {
			// 创建Consumer
			c, err = sub.client.CreateConsumer(sub.topicName, groupID, sub.messageTag)
			if err != nil {
				sub.logger.Panicf("startApache:CreateConsumer[%s][%s] err[%s]", sub.topicName, sub.messageTag, err)
				return
			}

			// 注册Subscriber
			err = sub.subscribeApache(c)
			if err != nil {
				sub.logger.Panicf("startApache:subscribeApache [%s][%s] err[%s]", sub.topicName, sub.messageTag, err)
				return
			}

			// Start consumer
			err = c.Start()
			if err != nil {
				sub.logger.Panicf("startApache:start Apache [%s][%s] err[%s]", sub.topicName, sub.messageTag, err)
				return
			}
		}
	}
}

func (se *SubscribeEngine) StartTrans() {
	if len(se.subTransList) == 0 {
		se.logger.Panicf("没有配置两阶段确认事务队列;")
	}
	// 二阶段消息监控
	for _, sub := range se.subTransList {
		se.logger.Infof("二阶段事务确认 Handler topicName[%s] groupID[%s]", sub.topicName, sub.groupID)

		if se.rocketMqType == internal.RocketMqApache { // 创建事务Producer
			sub.subscribeApacheTrans()
		} else {
			for i := 0; i < sub.concurrency; i++ {
				se.wg.Add(1)
				go func(sub *Subscriber) {
					se.logger.Infof("订阅信息: topicName[%s] groupID[%s] messageTag[%s]", sub.topicName, sub.groupID, sub.messageTag)
					for {
						select {
						case <-se.ctx.Done():
							se.logger.Infof("close the rocketmq sub")
							se.wg.Done()
							return
						default:
						}
						ctx, cancel := context.WithTimeout(se.ctx, 35*time.Second)
						sub.subscribeAliyunTrans(ctx)
						cancel()
					}
				}(sub)
			}
		}
	}

	se.logger.Infof("SubscribeEngine StartTrans Success!")
}

func (se *SubscribeEngine) Stop() {
	se.cancel()
	se.wg.Wait()
	se.logger.Infof("close all sub")
}
